package com.data.hijack.protocol.executor;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.nio.charset.Charset;
import java.util.Objects;

import org.apache.log4j.Logger;

import com.data.hijack.constant.Constants;
import com.data.hijack.context.HolderContext;
import com.data.hijack.readable.ReadablePlugin;
import com.data.hijack.util.Asserts;
import com.data.hijack.util.StringUtils;
import com.data.hijack.util.TextUtils;

/**
 * @author husq
 */
public class SocketExecutor extends AbstractSchemaExecutor {
    private static final Logger logger = Logger.getLogger(SocketExecutor.class);
    private volatile boolean readed;
    private SelectionKey selectionKey;

    @Override
    public void execute(SelectionKey selectionKey) throws Exception {
        this.selectionKey = selectionKey;
        this.testConnection();
        this.execute();
    }

    private void execute() throws Exception {
        String sendedMsg = "";
        if (Constants.IS_INTERCEPTABLE_1.equals(this.getSysConfig().getIsInterceptable())) {
            sendedMsg = this.getMsg();
        } else if (Constants.IS_INTERCEPTABLE_0.equals(this.getSysConfig().getIsInterceptable())) {
            String msg = "";
            ReadablePlugin readablePlugin = (ReadablePlugin)HolderContext.get(Constants.READABLE_PLUGIN_INSTANCE);
            if (!readed && Objects.nonNull(readablePlugin)) {
                msg = readablePlugin.read((SocketChannel)this.selectionKey.channel());
                this.readed = true;
            }
            if (StringUtils.isBlank(msg)) {
                msg = this.getMsg();
            }

            sendedMsg = "-1".equals(msg) ? msg : this.fire(msg);
        } else {
            Asserts.check(false, "Configuration error.");
        }

        if ("-1".equals(sendedMsg)) {
            return;
        }

        this.writeSocketChannel(this.selectionKey, sendedMsg);
    }

    private void testConnection() {
        int ccnt = this.getSysConfig().getMaxConnections();
        do {
            if (this.tryConnect()) {
                return;
            }
        } while (--ccnt > 0);

        String err = "The maximum number of attempts to connect has been exhausted";
        if (logger.isInfoEnabled()) {
            logger.info(TextUtils.format("%s[%s]", err, ccnt));
        }

        Asserts.check(false, err);
    }

    private String getMsg() {
        SocketChannel schannel = (SocketChannel)this.selectionKey.channel();
        final ByteBuffer buffer = ByteBuffer.allocate(8192);
        int size;
        try {
            size = schannel.read(buffer);
        } catch (IOException e) {
            String err = "Reading message exception";
            logger.error("Reading message exception", e);
            throw new RuntimeException(err);
        }

        if (size == -1) {
            this.closeChannel(schannel);
            this.closeChannel((SocketChannel)this.selectionKey.attachment());
            return "-1";
        }

        buffer.flip();
        String msg = Charset.forName(this.getSysConfig().getCharset()).decode(buffer).toString();

        return msg;
    }

    private boolean tryConnect() {
        SocketChannel schannel = (SocketChannel)this.selectionKey.channel();
        SocketChannel ochannel = (SocketChannel)this.selectionKey.attachment();
        try {
            if (Objects.isNull(ochannel)) {
                ochannel = SocketChannel.open();
                ochannel.configureBlocking(false);
                ochannel.register(this.selector, SelectionKey.OP_READ, schannel);
                schannel.register(this.selector, SelectionKey.OP_READ, ochannel);
            }
            if (!ochannel.isConnected()) {
                SocketAddress sa = new InetSocketAddress(this.getSysConfig().getHost(),
                    Integer.parseInt(this.getSysConfig().getPort()));
                ochannel.connect(sa);
                long time = System.currentTimeMillis();
                while (!ochannel.finishConnect()) {
                    if (this.determineTimeout(time)) {
                        return false;
                    }
                }
            }
        } catch (IOException e) {
            logger.error(TextUtils.format(e, "%s[%s:%s]", "Connecting to the server encountered an error",
                this.getSysConfig().getHost(), this.getSysConfig().getPort()));
            throw new RuntimeException("System error");
        } catch (Exception e) {
            logger.error(TextUtils.format(e, "%s[%s:%s]", "Connecting to the server encountered an error",
                this.getSysConfig().getHost(), this.getSysConfig().getPort()));
            throw new RuntimeException("System error");
        }

        return true;
    }

    private boolean determineTimeout(long time) {
        if (System.currentTimeMillis() - time > Integer.parseInt(this.getSysConfig().getTimeout())) {
            logger.warn(TextUtils.format("%s[%s:%s]", "Connection timed out.", this.getSysConfig().getHost(),
                this.getSysConfig().getPort()));
            return true;
        }

        return false;
    }

    private void writeSocketChannel(SelectionKey selectionKey, String msg) {
        if (StringUtils.isBlank(msg)) {
            logger.warn(TextUtils.format("%s[%s:%s]", "Message is empty", this.getSysConfig().getHost(),
                this.getSysConfig().getPort()));
            return;
        }

        final SocketChannel schannel = (SocketChannel)selectionKey.channel();
        if (!schannel.isOpen()) {
            return;
        }

        final SocketChannel ochannel = (SocketChannel)selectionKey.attachment();
        try {
            ochannel.write(ByteBuffer.wrap(msg.getBytes(this.getSysConfig().getCharset())));
        } catch (Exception e) {
            logger.error(e);
            throw new RuntimeException("System errror");
        }
    }
}
